package com.hery.demo;

import com.hery.functions.SourceFunctions;
import com.hery.utils.Config;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class MysqlCDCSource {

    public static void main(String[] args) throws Exception {

        Configuration config = Config.initApplicationConfig(args, MysqlCDCSource.class);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
/*
        env.enableCheckpointing(20, CheckpointingMode.AT_LEAST_ONCE);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(60 * 1000);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10,3000L));
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setStateBackend(new FsStateBackend("file:///qisi/gitlab/flink-real-time-alert/cp",false));
*/


        DataStreamSource<String> mysqlRules = env.addSource(SourceFunctions.getMysqlRuleSource());

//        SingleOutputStreamOperator<JSONObject> mysqlStream = mysqlRules;
//                .map(s -> {
//                    JSONObject jsonObject = JSON.parseObject(s).getJSONObject("after");
//                    System.out.println("jsonObject.getLongValue() = " + jsonObject.getLongValue("create_time"));
//                    System.out.println("jsonObject.getLong() = " + jsonObject.getLong("create_time"));
//                    return jsonObject;
//                })
//                .assignTimestampsAndWatermarks(
//                        WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3000))
//                                .withTimestampAssigner((ele, record) -> {
//                                    System.out.println("ele = " + ele);
//                                    try {
//                                        ele.getLong("create_time");
//                                    } catch (Exception e) {
//                                        System.out.println("ele = " + ele);
//                                        e.printStackTrace();
//                                    }
//                                    return 0;
//                                        }
//
//                                ))
//                ;

        mysqlRules.print();

        env.execute("mysql-cdc");
    }

}
